package cn.initcap;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author initcap
 * @date Created in 8/28/19 8:30 PM.
 */
public class StreamingFlinkWordCountJava {

    public static void main(String[] args) throws Exception {

        // 1. 获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 获取数据
        DataStreamSource<String> text = env.socketTextStream("localhost", 9999);

        // 3. 进行转换
        text.flatMap((String value, Collector<WordCount> collector) -> {
            String[] tokens = value.split(",");
            for (String token : tokens) {
                collector.collect(new WordCount(token, 1));
            }
        }).keyBy("word")
                .timeWindow(Time.seconds(5))
                .sum("count")
                .print()
                .setParallelism(1);

        // 4. 提交执行
        env.execute("StreamingFlinkWordCountJava");
    }

}
